package com.baishancloud.log.report

import com.baishancloud.log.common.util.DimMessageFetch
import com.baishancloud.log.format.scala.Origin
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.slf4j.{Logger, LoggerFactory}

import java.util.{Objects, Timer, TimerTask}
import scala.collection.mutable

/**
 * 根据获取到的最新域名和id过滤数据<br>
 * 域名半小时更新一次<br>
 *
 * @param parameterTool 入参
 * @author ziqiang.wang
 * @date 2021/11/18 14:38
 */
class DomainIdFilter(parameterTool: ParameterTool) extends RichFilterFunction[Origin] {

  private val LOGGER: Logger = LoggerFactory.getLogger(classOf[DomainIdFilter])
  private val domainId: mutable.Map[String, Int] = mutable.Map[String, Int]()
  private var subTaskIndex: Int = -1

  override def open(parameters: Configuration): Unit = {
    subTaskIndex = getRuntimeContext.getIndexOfThisSubtask
    DimMessageFetch.fetchAccountDomainId(parameterTool.get(accounts), domainId)
    LOGGER.info(s"DomainIdFilter${subTaskIndex}初始化完成，domainId表大小为：${domainId.size}")
    val timer: Timer = new Timer("update domainId", true)
    //增加定时器，每半小时获取一次最新域名id信息
    timer.scheduleAtFixedRate(new TimerTask {
      override def run(): Unit = DimMessageFetch.fetchAccountDomainId(parameterTool.get(accounts), domainId)
    }, 1800 * 1000, 1800 * 1000)
  }

  override def filter(value: Origin): Boolean = {
    domainId.contains(value.domain()) && value.httpCode() >= 500 && Objects.equals(value.hit(), "TCP_MISS") && Objects.equals(value.hierarchy(), "DIRECT")
  }
}
